package com.wxy.demo.rocketmq;

import cn.hutool.json.JSONUtil;
import com.wxy.demo.dto.StepDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.ArrayList;
import java.util.List;

/**
 * <p>
 *
 * </p>
 *
 * @author e-Xiaoyuan.Wang
 * @since 2022/11/7 17:15
 */
@Slf4j
public class ProducerDemo {
    public static void main(String[] args) throws Exception {

        //创建发送消息对象
        DefaultMQProducer producer = new DefaultMQProducer("group1", getAclRPCHook());
        //设定命名服务器地址---获取到消息服务器ip
        producer.setNamesrvAddr("127.0.0.1:9876");
        //启动发送服务
        producer.start();
        // 构建数据
        ArrayList<StepDto> list = new ArrayList<>();
        for (int i = 0; i < 2; i++) {
            String orderNo = "No." + i;
            for (int j = 0; j < 10; j++) {
                list.add(new StepDto().setOrderNo(orderNo).setSort(j));
            }
        }
        for (StepDto stepDto : list) {
            //构建消息，指定topic和body
            String topic = "topic1";
            String content = JSONUtil.toJsonStr(stepDto);
            Message msg = new Message(topic, content.getBytes());
            //发送消息
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    return list.get((stepDto.getOrderNo().hashCode() & Integer.MAX_VALUE) % list.size());
                }
            }, 10000);
            log.info("发送消息：status = {}, queue = {}, topic = {}, content = {}", sendResult.getSendStatus().name(),
                    sendResult.getMessageQueue().getQueueId(), topic, content);
        }
        //关闭连接
        producer.shutdown();
    }

    /**
     * 添加认证
     *
     * @return
     */
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("RocketMQ", "12345678"));
    }
}
